package com.rk.recommand;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.*;

public class UserCF_Step5 {
    public static class Step5Mapper extends Mapper<LongWritable, Text, Text, Text>{

        Text k = new Text();
        Text v = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] words = CommonConst.DELIMITER.split(line);
            if(words.length >= 3){
                k.set(words[0]);
                v.set(words[1] + "," + words[2]);
                context.write(k, v);
            }
        }
    }

    public static class Step5Reducer extends Reducer<Text, Text, Text, Text>{
        private final int RECOMMENDER_NUM = 10;  // 推荐数量

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            Map<Double, String> map = new HashMap<Double, String>();

            for (Text line : values) {
                String val = line.toString();
                String[] vlist = CommonConst.DELIMITER.split(val);

                if (vlist.length >= 2)
                {
                    map.put(Double.parseDouble(vlist[1]), vlist[0]);
                }
            }

            List<Double> list = new ArrayList<Double>();
            Iterator<Double> iter = map.keySet().iterator();
            while (iter.hasNext()) {
                Double similarity = iter.next();
                list.add(similarity);
            }

            //然后通过比较器来实现排序
            Collections.sort(list,new Comparator<Double>() {
                //降序排序
                public int compare(Double o1, Double o2) {
                    return o2.compareTo(o1);
                }
            });

            String v = "";
            for (int i = 0; i < RECOMMENDER_NUM && i < list.size(); i++)
            {
                if (list.get(i).compareTo(new Double(0.001)) > 0)
                {
                    // v += "," + map.get(list.get(i)) + "[" + String.format("%.2f", list.get(i)) + "]";
                    v += "," + map.get(list.get(i)) + ":" + String.format("%.2f", list.get(i));
                }
            }

            if (v != null && v.length() > 0)
            {
                context.write(key, new Text(v.substring(1)));
            }
            else
            {
                context.write(key, new Text("none"));
            }

        }
    }

    public static void run(Map<String, String> path) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        System.setProperty("HADOOP_USER_NAME", "root");
        conf.set("mapreduce.job.jar", "D:\\projects\\hadoop-study\\target\\hadoop-study-1.0-SNAPSHOT-jar-with-dependencies.jar");
        conf.set("mapreduce.app-submission.cross-platform", "true");//意思是跨平台提交

        String input  = path.get("input_step5");
        String output = path.get("output_step5");

        Job job = Job.getInstance(conf, "UserCF_Step5 job");
        job.setJarByClass(UserCF_Step5.class);
        job.setMapperClass(Step5Mapper.class);
        job.setReducerClass(Step5Reducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setOutputFormatClass(RedisOutputFormat.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.setInputPaths(job, new Path(input));// 加载2个输入数据集
        FileOutputFormat.setOutputPath(job, new Path(output));

        System.out.println("input : " + input);
        System.out.println("output: " + output);

        if (!job.waitForCompletion(true))
        {
            System.out.println("UserCF_Step5 run stop!");
            return;
        }

        System.out.println("UserCF_Step5 run successfully!");
    }

}
